//! Kafka集成使用示例
//! 
//! 展示如何使用Kafka消息队列进行沙箱事件处理和任务调度

use std::sync::Arc;
use std::collections::HashMap;
use anyhow::Result;
use tokio::time::{sleep, Duration};

use koda_sandbox_manager::{Config, Manager};
use koda_sandbox_manager::kafka::{
    KafkaManager, SandboxEventType, ErrorLevel, HealthStatus,
    MetricsType,
};
use koda_sandbox_manager::types::{Language, SandboxState, Priority};

#[tokio::main]
async fn main() -> Result<()> {
    // 初始化日志
    tracing_subscriber::init();
    
    println!("🚀 启动Kafka集成示例...");
    
    // 1. 初始化沙箱管理器
    let config = Config::default();
    let manager = Manager::new(config.clone()).await?;
    manager.start().await?;
    
    let sandbox_manager = manager.sandbox_manager();
    
    // 2. 初始化Kafka管理器
    let mut kafka_manager = KafkaManager::new(config.kafka.clone())
        .with_sandbox_manager(sandbox_manager);
    
    // 3. 初始化生产者
    kafka_manager.init_producer().await?;
    println!("✅ Kafka生产者初始化完成");
    
    // 4. 初始化消费者
    kafka_manager.init_consumer().await?;
    println!("✅ Kafka消费者初始化完成");
    
    // 5. 启动消费者（在后台运行）
    let kafka_manager_arc = Arc::new(kafka_manager);
    let consumer_manager = kafka_manager_arc.clone();
    
    tokio::spawn(async move {
        if let Err(e) = consumer_manager.start_consumer().await {
            eprintln!("❌ 消费者启动失败: {}", e);
        }
    });
    
    println!("✅ Kafka消费者启动完成");
    
    // 6. 发送各种消息示例
    println!("\n📨 开始发送消息示例...");
    
    // 发送沙箱事件
    let sandbox_id = uuid::Uuid::new_v4();
    let mut metadata = HashMap::new();
    metadata.insert("version".to_string(), serde_json::json!("1.0"));
    metadata.insert("region".to_string(), serde_json::json!("us-west-2"));
    
    kafka_manager_arc.send_sandbox_event(
        sandbox_id,
        SandboxEventType::Created,
        SandboxState::Creating,
        Language::Python,
        metadata,
    ).await?;
    println!("📦 发送沙箱事件: 沙箱创建");
    
    sleep(Duration::from_millis(500)).await;
    
    // 发送执行任务
    let task_id = format!("task_{}", uuid::Uuid::new_v4());
    let mut env_vars = HashMap::new();
    env_vars.insert("PYTHONPATH".to_string(), "/opt/app".to_string());
    
    kafka_manager_arc.send_execution_task(
        task_id.clone(),
        Some(sandbox_id),
        "print('Hello from Kafka task!')".to_string(),
        Language::Python,
        "main".to_string(),
        env_vars,
        "".to_string(),
        Priority::Normal,
        30000, // 30秒超时
    ).await?;
    println!("⚡ 发送执行任务: {}", task_id);
    
    sleep(Duration::from_millis(500)).await;
    
    // 发送系统指标
    let mut metrics = HashMap::new();
    metrics.insert("cpu_usage_percent".to_string(), 75.5);
    metrics.insert("memory_usage_percent".to_string(), 68.2);
    metrics.insert("active_sandboxes".to_string(), 12.0);
    
    let mut labels = HashMap::new();
    labels.insert("component".to_string(), "sandbox-manager".to_string());
    labels.insert("version".to_string(), "1.0.0".to_string());
    
    kafka_manager_arc.send_system_metrics(
        MetricsType::SystemResource,
        metrics,
        labels,
    ).await?;
    println!("📊 发送系统指标");
    
    sleep(Duration::from_millis(500)).await;
    
    // 发送错误报告
    let mut context = HashMap::new();
    context.insert("function".to_string(), serde_json::json!("execute_code"));
    context.insert("line".to_string(), serde_json::json!(123));
    
    kafka_manager_arc.send_error_report(
        ErrorLevel::Warning,
        "SANDBOX_TIMEOUT".to_string(),
        "沙箱执行超时".to_string(),
        Some("执行时间超过了30秒的限制".to_string()),
        Some(sandbox_id),
        Some("stack trace here...".to_string()),
        context,
    ).await?;
    println!("🚨 发送错误报告");
    
    sleep(Duration::from_millis(500)).await;
    
    // 发送健康检查结果
    let mut health_details = HashMap::new();
    health_details.insert("firecracker_version".to_string(), serde_json::json!("1.12.1"));
    health_details.insert("total_sandboxes".to_string(), serde_json::json!(15));
    health_details.insert("available_memory_mb".to_string(), serde_json::json!(2048));
    
    kafka_manager_arc.send_health_check(
        "sandbox-manager".to_string(),
        HealthStatus::Healthy,
        health_details,
        25, // 25ms响应时间
    ).await?;
    println!("💚 发送健康检查结果");
    
    // 7. 批量发送示例
    println!("\n📬 批量发送消息示例...");
    
    for i in 1..=5 {
        let task_id = format!("batch_task_{}", i);
        kafka_manager_arc.send_execution_task(
            task_id.clone(),
            None, // 自动分配沙箱
            format!("print('Batch task #{}')", i),
            Language::Python,
            "main".to_string(),
            HashMap::new(),
            "".to_string(),
            Priority::Low,
            15000,
        ).await?;
        
        println!("  📝 发送批量任务 {}: {}", i, task_id);
        sleep(Duration::from_millis(200)).await;
    }
    
    // 8. 刷新生产者缓冲区
    kafka_manager_arc.flush_producer(5000).await?;
    println!("✅ 生产者缓冲区刷新完成");
    
    // 9. 等待消息处理
    println!("\n⏳ 等待消息处理...");
    sleep(Duration::from_secs(10)).await;
    
    // 10. 停止消费者
    kafka_manager_arc.stop_consumer().await?;
    println!("🛑 Kafka消费者已停止");
    
    // 11. 停止沙箱管理器
    manager.stop().await?;
    println!("🛑 沙箱管理器已停止");
    
    println!("\n✅ Kafka集成示例完成！");
    println!("📋 本示例展示了以下功能:");
    println!("  • Kafka生产者和消费者的初始化");
    println!("  • 沙箱事件的发布和订阅");
    println!("  • 异步任务的分发和处理");
    println!("  • 系统指标的收集和传输");
    println!("  • 错误报告的统一处理");
    println!("  • 健康检查结果的监控");
    println!("  • 批量消息处理");
    println!("  • 优雅关闭和资源清理");
    
    Ok(())
}